package top.wintp.offlinedataanalysis.anlyser.mr;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.mapred.FileOutputCommitter;
import org.apache.hadoop.mapreduce.JobContext;
import org.apache.hadoop.mapreduce.OutputCommitter;
import org.apache.hadoop.mapreduce.OutputFormat;
import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.ibatis.session.SqlSession;
import org.apache.ibatis.session.SqlSessionFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.convert.Convert;
import top.wintp.offlinedataanalysis.anlyser.dao.StatsDeviceBrowserDAO;
import top.wintp.offlinedataanalysis.anlyser.dao.StatsUserDAO;
import top.wintp.offlinedataanalysis.anlyser.dim.StatsUserDimension;
import top.wintp.offlinedataanalysis.anlyser.domain.BaseTableObj;
import top.wintp.offlinedataanalysis.anlyser.domain.StatsDeviceBrowser;
import top.wintp.offlinedataanalysis.anlyser.domain.StatsUser;
import top.wintp.offlinedataanalysis.anlyser.value.reduce.MapWritableValue;
import top.wintp.offlinedataanalysis.common.ConstantValues;
import top.wintp.offlinedataanalysis.common.KpiType;
import top.wintp.offlinedataanalysis.util.MyBatisUtils;

/**
 * @author: pyfysf
 * <p>
 * @qq: 337081267
 * <p>
 * @CSDN: http://blog.csdn.net/pyfysf
 * <p>
 * @blog: http://wintp.top
 * <p>
 * @email: pyfysf@163.com
 * <p>
 * @time: 2019/9/6
 */
public class MySqlOutputFormat extends OutputFormat<StatsUserDimension, MapWritableValue> {
    private static final Logger logger = LoggerFactory.getLogger(MySqlOutputFormat.class);


    @Override
    public RecordWriter<StatsUserDimension, MapWritableValue> getRecordWriter(TaskAttemptContext context) throws IOException, InterruptedException {
        return new MySqlRecordWriter(context);
    }

    @Override
    public void checkOutputSpecs(JobContext context) throws IOException, InterruptedException {

    }

    @Override
    public OutputCommitter getOutputCommitter(TaskAttemptContext context) throws IOException, InterruptedException {
        return new FileOutputCommitter();
    }

    /**
     * 具体的输入对象
     */
    private class MySqlRecordWriter extends RecordWriter<StatsUserDimension, MapWritableValue> {
        private final TaskAttemptContext context;
        private final Configuration conf;
        private Writable valueKey = new IntWritable(-1);
        /**
         * 缓存
         */
        List<StatsUser> cacheStatsUsers = new ArrayList<>();
        /**
         * 缓存
         */
        List<StatsDeviceBrowser> cacheStatsBrowserUsers = new ArrayList<>();


        public MySqlRecordWriter(TaskAttemptContext context) {
            this.context = context;
            this.conf = context.getConfiguration();
        }

        @Override
        public void write(StatsUserDimension key, MapWritableValue value) throws IOException, InterruptedException {
            //获取值
            IntWritable resultCounts = (IntWritable) value.getValue().get(valueKey);
            //获取维度
            KpiType kpi = value.getKpi();
            //转化维度 如果维度存在获取id 不存在保存之后获取id
            int dateId = ConvertDimension.convertDateDimension(key.getStatsCommon().getDate());
            int platformId = ConvertDimension.convertPlatformDimension(key.getStatsCommon().getPlatform());
            int browserId = ConvertDimension.convertBrowserDimension(key.getBrowser());

            logger.info("MySqlRecordWriter  write()   kpi   " + kpi.name);
            //   判断是哪张表
            if (kpi.name.equalsIgnoreCase(KpiType.NEW_INSTALL_USER.name)) {
                StatsUser obj = new StatsUser();
                obj.setDateDimensionId(dateId);
                obj.setPlatformDimensionId(platformId);
                //新增用户
                obj.setNewInstallUsers(resultCounts.get());
                cacheStatsUsers.add(obj);
            } else if (kpi.name.equalsIgnoreCase(KpiType.BROWSER_NEW_INSTALL_USER.name)) {
                StatsDeviceBrowser obj = new StatsDeviceBrowser();
                obj.setDateDimensionId(dateId);
                obj.setPlatformDimensionId(platformId);
                obj.setBrowserDimensionId(browserId);
                //新增用户
                obj.setNewInstallUsers(resultCounts.get());
                cacheStatsBrowserUsers.add(obj);
            }

            insertData();
        }

        /**
         * 插入数据
         */
        private void insertData() {
            if (cacheStatsBrowserUsers.size() % this.conf.getInt(ConstantValues.MYSQL_BATCH_NUM_KEY, ConstantValues.DEFAULT_MYSQL_BATCH_NUM) == 0) {
                insertBrowserUser();
            }
            if (cacheStatsUsers.size() % this.conf.getInt(ConstantValues.MYSQL_BATCH_NUM_KEY, ConstantValues.DEFAULT_MYSQL_BATCH_NUM) == 0) {
                insertStatsUser();
            }
        }

        /**
         * 插入用户
         */
        private void insertStatsUser() {
            if (CollectionUtil.isEmpty(cacheStatsUsers)) {
                return;
            }

            SqlSessionFactory sqlSessionFactory = MyBatisUtils.getSqlSessionFactory();
            SqlSession sqlSession = sqlSessionFactory.openSession(false);
            //    执行插入语句
            StatsUserDAO mapper = sqlSession.getMapper(StatsUserDAO.class);
            mapper.batchInsertDataSelective(cacheStatsUsers);
            sqlSession.commit();
            cacheStatsUsers.clear();
            sqlSession.close();
        }

        /**
         * 插入浏览器用户
         */
        private void insertBrowserUser() {
            if (CollectionUtil.isEmpty(cacheStatsBrowserUsers)) {
                return;
            }


            SqlSessionFactory sqlSessionFactory = MyBatisUtils.getSqlSessionFactory();
            SqlSession sqlSession = sqlSessionFactory.openSession(false);
            //    执行插入语句
            StatsDeviceBrowserDAO mapper = sqlSession.getMapper(StatsDeviceBrowserDAO.class);
            mapper.batchInsertDataSelective(cacheStatsBrowserUsers);
            sqlSession.commit();
            cacheStatsBrowserUsers.clear();
            sqlSession.close();
        }

        @Override
        public void close(TaskAttemptContext context) throws IOException, InterruptedException {
            if (CollectionUtil.isNotEmpty(cacheStatsBrowserUsers)) {
                insertBrowserUser();
                cacheStatsBrowserUsers.clear();
            }
            if (CollectionUtil.isNotEmpty(cacheStatsUsers)) {
                insertStatsUser();
                cacheStatsUsers.clear();
            }
        }
    }
}
